package xin.alum.aim.coder;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.SneakyThrows;
import org.springframework.scheduling.annotation.Async;
import org.springframework.util.StopWatch;
import xin.alum.aim.constant.AimConstant;
import xin.alum.aim.constant.AimAttr;
import xin.alum.aim.model.Transportable;

import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

/**
 * 不能设置成Share 否则queue会成为单例而积压
 *
 * @param <T>
 */
abstract class BaseEncoder<T extends Transportable> extends MessageToMessageEncoder<T> {
    protected InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());

    private Comparator<T> comparator = (o1, o2) -> o1.getPriority() > o2.getPriority() ? o1.getPriority() : o2.getPriority();

    private PriorityBlockingQueue<T> queue = new PriorityBlockingQueue(256, comparator);

    public BaseEncoder() {

    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof Transportable) {
            queue.put((T) msg);
            send(ctx, promise);
        } else {
            super.write(ctx, msg, promise);
        }
    }

    @SneakyThrows
    @Async
    public void send(ChannelHandlerContext ctx, ChannelPromise promise) {
        int c_success = 0;
        StopWatch watch = new StopWatch();
        watch.start();
        while (queue.iterator().hasNext()) {
            T m = queue.poll();
            super.write(ctx, m, ctx.newPromise());
            c_success++;
        }
        watch.stop();
        if (queue.size() > AimConstant.CHANNEL_MSG_COUNT) {
            logger.warn("{}向用户【{}】写入【{}/{}】,用时【{}】", ctx.channel(), ctx.channel().attr(AimAttr.UID), c_success, queue.size(), watch.getTotalTimeMillis());
        }
    }
}
